package com.pk.flink.basic;


import com.pk.flink.basic.function.PKFlatMapFunction;
import com.pk.flink.basic.function.PKMapFunction;
import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * 使用Flink 实现单词计数
 * @author pk
 */
public class StreamBatchWordCountApp {
    public static void main(String[] args) throws Exception {
        //获取上下文环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setRuntimeMode(RuntimeExecutionMode.AUTOMATIC);
        //读取文件内容
        DataStreamSource<String> stringDataSource = env.readTextFile("data/wc_data", "UTF-8");

        //拆分行为单词数组
        stringDataSource.flatMap(new PKFlatMapFunction()) //   aaa,bbb,ccc  =>  ["aaa","aaa","bbb"]
                        .map(new PKMapFunction()) // ["aaa","aaa","bbb"] => [("aaa",1),("aaa",1),("bbb",1)]
                        .keyBy(x->x.f0) // [("aaa",1),("aaa",1),("bbb",1)] => [("aaa",[1,1]),("bbb",[1])]
                        .sum(1)
                        .print();
        env.execute();

        System.out.println("=================================");
        stringDataSource.flatMap((String s, Collector<String> collector) -> {
                            String[] words = s.split(",");
                            for (String word : words) {
                                collector.collect(word);
                            }
                         }).returns(Types.STRING) //   aaa,bbb,ccc  =>  ["aaa","aaa","bbb"]
                        .map((String word) -> new Tuple2<>(word, 1)) // ["aaa","aaa","bbb"] => [("aaa",1),("aaa",1),("bbb",1)]
                        .returns(Types.TUPLE(Types.STRING, Types.INT))
                        .keyBy(x->x.f0) // [("aaa",1),("aaa",1),("bbb",1)] => [("aaa",[1,1]),("bbb",[1])]
                        .sum(1)
                        .print();
        env.execute();
    }
}
